package com.cndw.kungfu.core;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import com.cndw.asyncsync.DataSyncServer;
import com.cndw.asyncsync.SimpleThreadFactory;
import com.cndw.asyncsync.SyncServerBase;
import com.cndw.kungfu.domain.LogPlayer;

public class LogSyncServer extends SyncServerBase {

	private ConcurrentLinkedQueue<LogPlayer> syncQueue = new ConcurrentLinkedQueue<LogPlayer>();

	private static LogSyncServer instance = null;

	private ExecutorService executor;

	private SyncProcess syncProcess;

	/**
	 * 每次处理多少条
	 */
	private int step = 50;
	/**
	 * 每分钟处理多少次
	 */
	private int speed = 60000;

	private boolean serverStop = false;

	public static LogSyncServer getInstance() {
		if (instance == null) {
			synchronized (DataSyncServer.class) {
				if (instance == null) {
					instance = new LogSyncServer();
				}
			}
		}
		return instance;
	}

	private LogSyncServer() {
		syncProcess = new SyncProcess();
	}

	public void commitSync(LogPlayer data) {
		syncQueue.add(data);
	}

	public int size() {
		return syncQueue.size();
	}

	public int getStep() {
		return step;
	}

	public int getSpeed() {
		return speed;
	}

	public synchronized void setSpeedAndStep(int speed, int step) {
		speed = speed <= 0 ? 100 : speed;
		step = step <= 0 || step >= 500 ? 500 : step;
		this.speed = speed;
		this.step = step;
		if (executor == null) {
			executor = Executors.newSingleThreadExecutor(new SimpleThreadFactory("Log Sync"));
		}
		if (syncProcess != null) {
			syncProcess.notifySpeed();
			if (!syncProcess.isAlive()) {
				syncProcess.start();
			}
		}
	}

	public void shutDown() {
		logger.info("log sync shutdown ... ...");
		serverStop = true;
		try {
			syncProcess.join();
		} catch (Exception e) {
			e.printStackTrace();
		}

		try {
			if (executor != null) {
				executor.shutdown();
				executor.awaitTermination(300, TimeUnit.SECONDS);
			}

		} catch (InterruptedException e) {
			e.printStackTrace();
		}

		logger.info("log sync stop suc");
	}

	private class SyncProcess extends Thread {
		private int millisWait = 200;

		public SyncProcess() {
			setName("log Sync ");
		}

		public void notifySpeed() {
			millisWait = 1000 * 60 / (speed / step);
			logger.info("log interval " + millisWait + " millisecond,speed:" + speed + ",step:" + step);
		}

		@Override
		public void run() {

			do {

				try {

					initTask();

					if (serverStop) {
						if (syncQueue.isEmpty()) {
							break;
						}
					} else {
						Thread.sleep(millisWait);
					}
				} catch (Exception e) {
					e.printStackTrace();
				}
			} while (true);

		}

		private void initTask() {
			
			if (stopOnDBErr) {
				return ;
			}
			
			final List<LogPlayer> logBatchObjects = new ArrayList<LogPlayer>();
			for (int i = 0; i < step && !syncQueue.isEmpty(); i++) {
				LogPlayer logData = syncQueue.remove();
				logBatchObjects.add(logData);
			}
			if (!logBatchObjects.isEmpty()) {
				Runnable r = new Runnable() {
					@Override
					public void run() {
						LogPlayer.save(logBatchObjects);
					}
				};
				executor.execute(r);
			}

		}
	}

}
